package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo4Burk {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("burl")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取数据
    val burks: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING , year STRING , tsl01 DOUBLE , tsl02 DOUBLE , tsl03 DOUBLE , tsl04 DOUBLE , tsl05 DOUBLE , tsl06 DOUBLE , tsl07 DOUBLE , tsl08 DOUBLE , tsl09 DOUBLE , tsl10 DOUBLE , tsl11 DOUBLE , tsl12 DOUBLE")
      .load("data/burks.txt")

    burks.createOrReplaceTempView("burks")


    /**
      * 1、统计每个公司每年按月累计收入  行转列 --> sum窗口函数
      *
      * 输出结果
      * 公司代码,年度,月份,当月收入,累计收入
      *
      */

    spark.sql(
      """
        |select explode(Array(1,2,3,4,5,6,7))
        |
      """.stripMargin)
      .show()


    spark.sql(
      """
        |select explode(map('001',23,'002',24,'003',25))
        |
      """.stripMargin).show()

    /**
      * sum  over     只分区，全局求和
      * sum  over     分区右排序，累计求和
      *
      */

    spark.sql(
      """
        |select
        | burk,year,month,pic,
        | sum(pic) over(partition by burk,year order by month) as sumPic,
        | sum(pic) over(partition by burk,year ) as sumP
        |from (
        |select burk,year,month,pic
        |from burks
        |lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) T as month,pic
        |) as a
      """.stripMargin).show()


    val mapCol: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )


    val burksMonthDF: DataFrame = burks
      //将一行转换多行
      .select($"burk", $"year", explode(mapCol) as Array("month", "pic"))


    //计算收入按月累计
    burksMonthDF
      .withColumn("sumPic", sum($"pic") over Window.partitionBy($"burk", $"year").orderBy("month"))
      .show()


    /**
      * 2、统计每个公司当月比上年同期增长率  行转列 --> lag窗口函数
      * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
      *
      * coalesce: 返回第一个不为空的列
      *
      */
    burksMonthDF
      //1\获取上一年当月的收入金额
      .withColumn("lagPic", lag($"pic", 1, 0.0) over Window.partitionBy($"burk", $"month").orderBy($"year"))
      //计算增长率
      .withColumn("p", coalesce($"pic" / $"lagPic" - 1, expr("1.0")))

      .show()
  }

}
